// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jaewon

#include "coin2/base/request/session.h"

#include <deque>
#include <utility>

#include <boost/beast/version.hpp>
#include <glog/logging.h>

namespace coin2::base::request {

constexpr int kVersion = 11;

Session::Session(
    boost::asio::io_context* io_ctx,
    Callback* callback,
    std::shared_ptr<detail::SimpleObjectPool<RequestContext>> req_ctx_pool,
    ssl::context* ssl_ctx,
    const std::string& host,
    int port,
    const std::string& bind_ip,
    const std::string& http_proxy)
    : io_ctx_(io_ctx),
      callback_(callback),
      req_ctx_pool_(std::move(req_ctx_pool)),
      ssl_ctx_(ssl_ctx),
      host_(host),
      port_(port),
      bind_ip_(bind_ip),
      http_proxy_{http_proxy} {
  connection_ = std::make_shared<HttpsConnection>(
      host_,
      port_,
      bind_ip_,
      this,
      io_ctx_,
      ssl_ctx_,
      http_proxy_);
  connection_->Start();
}

Session::Session(
    boost::asio::io_context* io_ctx,
    Callback* callback,
    std::shared_ptr<detail::SimpleObjectPool<RequestContext>> req_ctx_pool,
    ssl::context* ssl_ctx,
    const std::string& host,
    int port,
    const std::string& http_proxy)
    : Session(io_ctx, callback, std::move(req_ctx_pool), ssl_ctx, host, port, "", http_proxy) {}

Session::~Session() {
  AsyncShutdown();
  if (req_) {
    req_.reset();
  }
}

void Session::Connect() { connection_->Start(); }

void Session::AsyncHead() {
  RequestType req;
  std::string_view target("/");

  auto callback = [this](boost::system::error_code ec, const SessionResponse* resp) {
    DLOG(INFO) << "session: " << this;
    if (resp->raw_request) {
      DLOG(INFO) << "req: \n" << *(resp->raw_request);
    }
    if (resp->raw_response) {
      DLOG(INFO) << *(resp->raw_response);
    }
  };

  req.version(kVersion);
  req.method(http::verb::head);
  req.keep_alive(true);
  req.target(boost::beast::string_view(target.data(), target.size()));
  req.set(http::field::host, host_);

  DLOG(INFO) << "Head request is sent to create live connection: \n" << req;
  AsyncRequest(std::move(req), std::move(callback));
}

void Session::AsyncRequest(std::string_view target, CallbackType callback) {
  RequestType req;
  req.version(kVersion);
  req.method(http::verb::get);
  req.keep_alive(true);
  req.target(boost::beast::string_view(target.data(), target.size()));
  req.set(http::field::host, host_);

  AsyncRequest(std::move(req), std::move(callback));
}

void Session::AsyncRequest(RequestType req, CallbackType callback) {
  SanitizeRequest(&req);
  auto now = GetCurrentTimestamp();
  last_request_ts_ = now;
  req_ = req_ctx_pool_->Alloc(std::move(req), std::move(callback), now);
  connection_->AsyncSend(&req_->request);
}

void Session::AsyncShutdown() {
  // 20220917(daniel): do not change error code if it is set
  if (req_ && req_->ec.value() == boost::system::errc::success /* ec defualt is success */) {
    req_->ec = boost::asio::error::basic_errors::operation_aborted;
  }
  if (connection_) {
    connection_->AsyncShutdown();
  }
}

void Session::OnConnected(std::weak_ptr<HttpsConnection> conn) {
  // Do nothing
}

void Session::OnRequestSent(std::weak_ptr<HttpsConnection> conn, const RequestType* raw_request) {
  // Do nothing
}

void Session::OnResponseReceived(
    std::weak_ptr<HttpsConnection> conn,
    const ResponseType* raw_response) {
  last_active_ts_ = GetCurrentTimestamp();
  int32_t http_code = 0;
  if (req_->ec.value() != boost::system::errc::success /* ec defualt is success */) {
    // session is timeout or shutdown
    http_code = req_->ec.value();
    req_->response.raw_response = raw_response;
  } else {
    // Ok
    req_->ec = boost::system::error_code();
    req_->response.raw_response = raw_response;
    http_code = raw_response->result_int();
  }
  req_->response.response_ts = last_active_ts_;

  std::string req_method =
      std::string(req_->request.method_string().data(), req_->request.method_string().size());
  std::string req_target =
      std::string(req_->request.target().data(), req_->request.target().size());

  req_.reset();
  callback_->OnSessionReady(
      shared_from_this(),
      (last_active_ts_ - last_request_ts_),
      req_method,
      req_target,
      http_code);
}

void Session::OnError(std::weak_ptr<HttpsConnection> conn, boost::system::error_code ec) {
  LOG(INFO) << "Error while connecting to " << host_ << ":" << port_ << " " << ec.message();
  if (req_ && req_->ec.value() == boost::system::errc::success) {
    req_->ec = ec;
  } else {
    // session is timeout or shutdown
  }
  callback_->OnSessionError(shared_from_this());
}

void Session::SanitizeRequest(RequestType* req) {
  if (req->find(http::field::version) == req->end()) {
    req->version(kVersion);
  }

  req->keep_alive(true);

  {
    auto it = req->find(http::field::host);
    if (it == req->end()) {
      req->set(http::field::host, host_);
    } else if (it->value() != host_) {
      throw std::runtime_error("invalid host");
    }
  }
}

bool Session::ProcessTimeout(int64_t timeout_start_ts) {
  if (last_request_ts_ > timeout_start_ts) {
    return false;
  }
  if (req_) {
    req_->ec = boost::system::errc::make_error_code(boost::system::errc::timed_out);
  }
  return true;
}

}  // namespace coin2::base::request
